package com.example.mq.mqserver;

import com.example.mq.common.*;
import com.example.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 这个 BrokerServer 就是咱们消息队列的本体服务器
 * 本质上就是个 TCP 的服务器
 */
public class BrokerServer {

    private ServerSocket serverSocket=null;

    //当前考虑一个 BrokerServer 上只有一个虚拟主机
    private VirtualHost virtualHost=new VirtualHost("default");

    //使用这个哈希表 表示当前所以会话(也就是有那些客户端 正在和服务器通信)
    //次数的 key 是 channelId, value 为对应的 Socket 对象
    private ConcurrentHashMap<String,Socket> sessions=new ConcurrentHashMap<String,Socket>();

    //引入线程池 来处理多个客户端的请求
    private ExecutorService executorService=null;

    //引入一个 boolean 变量来控制服务器是否继续运行
    private volatile boolean runnable=true;

    public BrokerServer(int port) throws IOException {
        serverSocket=new ServerSocket(port);
    }

    //启动服务器
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动服务器!");
        executorService= Executors.newCachedThreadPool();
        try {
            while (runnable){
                //获取客户端链接
                Socket clientSocket = serverSocket.accept();
                //把处理连接的逻辑 丢给线程池
                executorService.submit(()->{
                    processConnection(clientSocket);
                });
            }
        }catch (SocketException e){
            System.out.println("[BrokerServer] 服务器停止运行");
        }
    }
    //一般来说停止服务器 就是之间kill掉对应进程即可
    //此处还是单独搞一个停止方法
    public void stop() throws IOException {
        runnable=false;
        //把线程池中的任务都放弃 同时销毁线程
        executorService.shutdownNow();
        serverSocket.close();
    }

    //通过这个方法 来处理一个客户的连接
    //在一个连接中 可能会涉及到多个请求和响应
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream=clientSocket.getInputStream();
             OutputStream outputStream=clientSocket.getOutputStream()){
            //这里我们需要按照约定的格式来读取和解析,此时就需要用到 DataInputStream 和DataOutputStream
            try (DataInputStream dataInputStream=new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream=new DataOutputStream(outputStream)){
                while (true){
                    //1.读取请求并解析
                    Request request=readRequest(dataInputStream);
                    //2.根据请求计算响应
                    Response response=process(request,clientSocket);
                    //3.把响应写回给客户端
                    whiteResponse(dataOutputStream,response);
                }
            }catch (EOFException |SocketException e){
                //对于这个代码来说 DataInputStream 如果读到 EOF 就会抛出 EOFException 这个异常
                //此处我们就需要借助这个异常来结束循环
                System.out.println("[BrokerServer] connection 关闭 客户端的地址:"+clientSocket.getInetAddress().toString()+":"+clientSocket.getPort());
            }
        } catch (IOException |ClassNotFoundException | MqException e) {
            System.out.println("[BrokerServer] connection 出现异常!");
            e.printStackTrace();
        }finally {
            try {
                //当连接处理完了 就需要关闭 socket
                clientSocket.close();
                //一个TCP连接中 可能包含多个 channel 需要把当前这个 socket 对应的所有的 channel 也清理掉
                clearClosedSession(clientSocket);
            }catch (IOException e){
                e.printStackTrace();
            }
        }
    }


    //解析请求 转换成代码的对象返回
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request=new Request();
        //读取出四个字节的type
        request.setType(dataInputStream.readInt());
        //再读取四个字节的length
        request.setLength(dataInputStream.readInt());
        byte[] payload=new byte[request.getLength()];
        int n=dataInputStream.read(payload);
        if(n!=request.getLength()){
            throw  new IOException("读取请求格式出错");
        }
        request.setPayload(payload);
        return request;
    }

    //将响应写回
    private void whiteResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        //写入四个字节的type
        dataOutputStream.writeInt(response.getType());
        //写入四个字节的length
        dataOutputStream.writeInt(response.getLength());
        //写入数据部分
        dataOutputStream.write(response.getPayload());
        //刷新缓冲区!! 快速进入网卡 不会一致保存在内存中
        dataOutputStream.flush();
    }

    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        //处理一次请求 返回一次响应
        //1.把 request 中的 payload 做一个初步的解析
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        System.out.println("[Request] rid="+basicArguments.getRid()+", channelId="+basicArguments.getChannelId()+", type="+"" +
                request.getType() +", length="+request.getLength() );
        //2.根据 type 的值 进一步区分 这次请求要干啥
        boolean ok=true;
        if(request.getType()==0x1){
            //创建 channel
            sessions.put(basicArguments.getChannelId(),clientSocket);
            System.out.println("[BrokerServer] 创建 channel 完成 channelId="+basicArguments.getChannelId());
        }else if(request.getType()==0x2){
            //销毁 channel
            sessions.remove(basicArguments.getChannelId());
            System.out.println("[BrokerServer] 删除 channel 完成 channelId="+basicArguments.getChannelId());
        }else if(request.getType()==0x3){
            //创建交换机 此时 payload 就是 ExchangeDeclareArguments 对象
            ExchangeDeclareArguments arguments=(ExchangeDeclareArguments) basicArguments;
            //执行方法
            ok=virtualHost.exchangeDeclare(arguments.getExchangeName(),arguments.getExchangeType(),arguments.isDurable(),arguments.isAutoDelete(),arguments.getArguments());

        }else if(request.getType()==0x4){
            //销毁交换机
            ExchangeDeleteArguments arguments=(ExchangeDeleteArguments) basicArguments;
            ok=virtualHost.exchangeDelete(arguments.getExchangeName());
        }else if(request.getType()==0x5){
            //创建队列
            QueueDeclareArguments arguments= (QueueDeclareArguments) basicArguments;
            ok=virtualHost.queueDeclare(arguments.getQueueName(),arguments.isDurable(),arguments.isExclusive(),arguments.isAutoDelete(),arguments.getArguments());

        }else if(request.getType()==0x6){
            //删除队列
            QueueDeleteArguments arguments= (QueueDeleteArguments) basicArguments;
            ok=virtualHost.queueDelete(arguments.getQueueName());
        }else if(request.getType()==0x7){
            //创建绑定
            QueueBindArguments arguments= (QueueBindArguments) basicArguments;
            ok=virtualHost.queueBind(arguments.getExchangeName(),arguments.getQueueName(),arguments.getBindingKey());

        }else if(request.getType()==0x8){
            //删除绑定
            QueueUnbindArguments arguments= (QueueUnbindArguments) basicArguments;
            ok=virtualHost.queueUnbind(arguments.getQueueName(),arguments.getExchangeName());
        }else if(request.getType()==0x9){
            //发送消息
            BasicPublishArguments arguments= (BasicPublishArguments) basicArguments;
            ok=virtualHost.basicPublish(arguments.getExchangeName(),arguments.getRoutingKey(),arguments.getBasicProperties(),arguments.getBody());

        }else if(request.getType()==0xa){
            //订阅消息
            BasicConsumeArguments arguments= (BasicConsumeArguments) basicArguments;
            ok=virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(),
                    new Consumer() {
                        @Override
                        public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                            //先知道 当前收到的消息 要发给那个客户端
                            //1.此处的 ConsumerTag 其实就是 channelId 根据 channelId 去 sessions 中查询 就能得到对应的 socket对象
                            //就可以给其发送数据了
                            Socket clientSocket = sessions.get(consumerTag);
                            if(clientSocket==null||clientSocket.isClosed()){
                                throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭");
                            }
                            //2.构造响应数据
                            Response response=new Response();
                            SubScribeReturns subScribeReturns=new SubScribeReturns();
                            //0xc 表示服务器给消费者客户端推送的消息数据
                            response.setType(0xc);
                            //此处 response 的payload 其实就是一个 SubScribeReturns 对象
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(basicProperties);
                            subScribeReturns.setBody(body);
                            subScribeReturns.setChannelId(consumerTag);
                            //这里的 rid 是一次请求 对应一次响应 但是此时我们是回调方法 并没有客户端的请求 所以不设置也没关系
                            subScribeReturns.setRid("");
                            subScribeReturns.setOk(true);
                            byte[] payload=BinaryTool.toBytes(subScribeReturns);
                            response.setLength(payload.length);
                            response.setPayload(payload);
                            //3.写入到响应之中 此处的 dataOutputStream 不能 close
                            // 如果关闭了 就会直接把 clientSocket 里的 OutputStream 也关了
                            // 此时就无法继续给socket 后续写入了
                            DataOutputStream dataOutputStream=new DataOutputStream(clientSocket.getOutputStream());
                            whiteResponse(dataOutputStream,response);
                        }
                    });
        }else if(request.getType()==0xb) {
            //调用 basicAck 确认消息
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
        }else {
            //当前的type是非法的
            throw new MqException("[BrokerServer] 未知的 type! type="+request.getType());
        }
        //3.构造响应
        BasicReturns basicReturns=new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);
        byte[] payload=BinaryTool.toBytes(basicReturns);
        Response response=new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[Response] rid="+basicReturns.getRid()+", channelId="+basicReturns.getChannelId()+", type="+response.getType()+
                ", length="+response.getLength());
        return response;
    }

    private void clearClosedSession(Socket clientSocket) {
        // 这里要做的是 遍历上述的哈希表 找到value为 socket(该关闭的socket) 连接 对应的channel清理 也就是删除掉
        List<String> toDeleteChannelId=new ArrayList<>();
       for(Map.Entry<String,Socket> entry :sessions.entrySet()){
           if(entry.getValue()==clientSocket){
               //这里不能直接删除 不能一边遍历 一边删除哈希表的元素
               //把要删除的结果都存储起来
               toDeleteChannelId.add(entry.getKey());
           }
       }
       for (String channelId : toDeleteChannelId){
           sessions.remove(channelId);
           System.out.println("[BrokerServer] 清除sessions 完成! 被清理的 channelId="+channelId);
       }
    }

}
